package com.example.service;

import com.example.dto.User;
import io.fury.Fury;
import io.fury.config.Language;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.queue.DistributedDelayQueue;
import org.apache.curator.framework.recipes.queue.QueueConsumer;
import org.apache.curator.framework.recipes.queue.QueueSerializer;
import org.apache.curator.framework.state.ConnectionState;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;

@Service
public class DistributedDelayQueueServiceImpl implements DistributedDelayQueueService, InitializingBean, DisposableBean {

    public static final String PATH = "/zookeeper/delay";

    Logger logger = LoggerFactory.getLogger(getClass());

    private DistributedDelayQueue distributedDelayQueue;

    @Resource
    private ZookeeperService zookeeperService;

    @Override
    public DistributedDelayQueue getDistributedDelayQueue() {
        return distributedDelayQueue;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        this.distributedDelayQueue = zookeeperService.getDistributedDelayQueue(new QueueConsumer() {
            @Override
            public void stateChanged(CuratorFramework client, ConnectionState newState) {

            }

            @Override
            public void consumeMessage(Object message) throws Exception {
            	if (message instanceof User) {
            		 logger.warn("DistributedDelayQueueServiceImpl :" + ((User)message).getId());
				}
               
            }
        }, new QueueSerializer() {

            private Fury fury;

            public Fury getFury() {
                if (this.fury == null) {
                    this.fury = Fury.builder().withLanguage(Language.JAVA).requireClassRegistration(true).build();
                    this.fury.register(User.class);
                }
                return this.fury;
            }
            @Override
            public byte[] serialize(Object item) {
                return getFury().serialize(item);
            }

            @Override
            public Object deserialize(byte[] bytes) {
                return getFury().deserialize(bytes);
            }
        }, PATH);
        this.distributedDelayQueue.start();
    }

    @Override
    public void destroy() throws Exception {
        this.distributedDelayQueue.close();
    }
}
